package com.shujia.airPM25;

import com.shujia.tuijian.TuiJianDemo;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/*
    part-r-00000:
        date: 20180927-城市编号:1001	#66
        date: 20180927-城市编号:1002	#60
        date: 20180927-城市编号:1003	#63

        1001 #66

    city:
        1001,$万寿西宫,北京
        1002,$定陵,北京
        1003,$东四,北京

    (1001,[#66,$万寿西宫-北京])

    1001-万寿西宫-北京 66


 */
import java.io.IOException;

class MyMapper2 extends Mapper<LongWritable, Text, Text, Text> {
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        //part-r-00000
        //city
        //获取当前行数据的输入文件的名称
        FileSplit fs = (FileSplit) context.getInputSplit();
        String name = fs.getPath().getName();

        //将当前行数据转成java中String字符串类型
        String line = value.toString();

        //可能我这一行数据无效
        if (line == null || line.equals("")) return;


        //判断一下该行数据是从哪个文件过来
        if (name.contains("part-r")) {
            //根据指定的分隔符进行分割  空格
            String[] strings = line.split("\\s+");
            //如果到这里了，说明该行数据是从第一张表过来的
            //在城市前面添加一个#号，跟第二表的数据进行区分
            String string = strings[1];
            //再以冒号进行分割
            String cityId = string.split(":")[1];
            context.write(new Text(cityId), new Text("#" + strings[2]));
        } else if (name.contains("city")) {
            //如果到这里了，说明该行数据是从第二张表过来的
            String[] strings = line.split(",");
            //获取城市id
            String cityId = strings[0];
            String cityInfo = "$" + strings[1] + strings[2];
            context.write(new Text(cityId), new Text(cityInfo));
        }
    }
}

class MyReducer2 extends Reducer<Text, Text, Text, Text> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        //(1001,[#66,$万寿西宫-北京])

        StringBuffer sb = new StringBuffer();

        //对value值进行切分
        for (Text value : values) {
            String s = value.toString();
            if (s.startsWith("#")) {
                String s1 = s.substring(1);
                sb.append(s1).append(" ");
            } else {
                String s1 = s.substring(1);
                sb.append(s1).append(" ");
            }
        }
        context.write(key, new Text(sb.toString()));

    }
}


public class Pm25AndCity {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(Pm25AndCity.class);
        job.setNumReduceTasks(1);
        job.setJobName("与city表进行拼接");

        job.setMapperClass(MyMapper2.class);
        job.setReducerClass(MyReducer2.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.waitForCompletion(true);
    }
}
